#include "net/socket.h"
#include "net/processor.h"
#include "net/tcp_stream.h"
#include "net/poller.h"
#include "net/poll_thread_pool.h"
#include "net/acceptor.h"
#include "net/tcp_server.h"

namespace net {

TcpServer::TcpServer(Poller* base_poll, const std::string& ip,
                     const uint16 port)
    : base_poller_(base_poll), ip_(ip), port_(port) {
}

TcpServer::~TcpServer() {
  MapClear(&stream_map_);

  poll_thread_pool_.reset();
  base_poller_.reset();
}

void TcpServer::SetProcessor(Processor* processor) {
  processor_.reset(processor);
}

void TcpServer::Start() {
  if (poll_thread_pool_.get() != NULL) return;

  acceptor_.reset(new Acceptor(base_poller_.get(), this));
  if (!acceptor_->Listen(Addr(ip_, port_)) || !acceptor_->EnableRead()) {
    acceptor_.reset();
    return;
  }

  poll_thread_pool_.reset(new PollThreadPool(base_poller_.get()));
  poll_thread_pool_->Start();
}

TcpStream* TcpServer::AddConnection(Socket* sock) {
  Poller* poller = poll_thread_pool_->getPoller();
  CHECK(poller != NULL && sock != NULL);

  scoped_ptr<TcpStream> stream(new TcpStream(sock, poller));
  stream->SetProcessor(processor_.get());
  Closure* c = NewMethordClosure1(this, &TcpServer::RemoveConnection,
                                  stream.get());
  stream->SetCloseClosure(c);
  if (!stream->Init()) {
    return NULL;
  }
  ScopedMutex l(&mutex_map_);
  stream_map_.insert(std::pair<int32, TcpStream*>(stream->Fd(), stream.get()));

  return stream.release();
}

void TcpServer::RemoveConnection(void* stream) {
  scoped_ptr<TcpStream> stream_ptr((TcpStream*) stream);
  ScopedMutex l(&mutex_map_);
  stream_map_.erase(stream_ptr->Fd());
}
}
